package com.etlmaster.rowprocessor.steps;

import java.io.IOException;
import java.util.Map;

import org.apache.hadoop.hbase.client.HTable;

import com.etlmaster.executor.trans.core.Line;
import com.etlmaster.executor.utils.DataCollectionUtils;
import com.etlmaster.hbase.rowProcess.HbaseUtil_rowProcess;
import com.etlmaster.rowprocessor.core.AbstractWriter;

public class HbaseWriter extends AbstractWriter {

	private String tableName;
	private HTable htable;

	public HbaseWriter(String id, String jobId, String name, String code,
			Map<String, Object> paramMap, Map<String, Object> var) {
		super(id, jobId, name, code, paramMap, var);
		// TODO Auto-generated constructor stub
	}

	@Override
	public void read(Line line) {
		String day_id = DataCollectionUtils
				.getCurrentTimestampByFormate("yyyy-MM-dd");

		String newRowKey = "";
		newRowKey = HbaseUtil_rowProcess.getMaxRowKey(htable, day_id);
		HbaseUtil_rowProcess.put(htable, line, newRowKey);

	}

	@Override
	protected void init() {
		try {
			htable = new HTable(HbaseUtil_rowProcess.getConf(), tableName);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

	}

	@Override
	protected void stop() {
		HbaseUtil_rowProcess.closeTable(htable);
	}

}
